package com.orderpay_detect

import com.orderpay_detect.bean.{OrderEvent, ReceiptEvent}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, _}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

/**
  * @Description: TODO QQ1667847363
  * @author: xiao kun tai
  * @date:2021/12/4 20:37
  *                 双流实时对账，使用intervalJoin
  */
object TxMatchWithJoin {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) //定义事件时间语义

    //从文件中读取数据，并转换成样例类,提取时间戳生成watermark
    val receiptPath: String = "OrderPayDetect/src/main/resources/ReceiptLog.csv"
    val orderPath: String = "OrderPayDetect/src/main/resources/OrderLog.csv"
    //读取支付数据
    val receiptEventStream: KeyedStream[ReceiptEvent, String] = env
      .readTextFile(receiptPath).map(data => {
      val arr: Array[String] = data.split(",")
      ReceiptEvent(arr(0), arr(1), arr(2).toLong)
    }).assignAscendingTimestamps(_.timestamp * 1000L)
      .keyBy(_.txId)
    //读取订单数据
    val orderEventStream: KeyedStream[OrderEvent, String] = env
      .readTextFile(orderPath).map(data => {
      val arr: Array[String] = data.split(",")
      OrderEvent(arr(0).toLong, arr(1), arr(2), arr(3).toLong)
    })
      .filter(_.eventType == "pay")
      .assignAscendingTimestamps(_.timestamp * 1000L)
      .keyBy(_.txId)

    val resultStream: DataStream[(OrderEvent, ReceiptEvent)] = orderEventStream.intervalJoin(receiptEventStream)
      .between(Time.seconds(-3), Time.seconds(5))
      //        .lowerBoundExclusive()
      //        .upperBoundExclusive()
      .process(new TxMatchWithJoinResult)

    resultStream.print("matched")

    env.execute("tx match job")

  }

  class TxMatchWithJoinResult extends ProcessJoinFunction[OrderEvent,ReceiptEvent,(OrderEvent,ReceiptEvent)]{
    override def processElement(in1: OrderEvent, in2: ReceiptEvent, context: ProcessJoinFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)]#Context, collector: Collector[(OrderEvent, ReceiptEvent)]): Unit = {
      collector.collect((in1,in2))
    }
  }


}
